package com.millstein;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class JoinTest4 {

    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 5000);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

//        tableEnv.executeSql(
//                "create table t34 (\n" +
//                "    id string,\n" +
//                "    name string,\n" +
//                "    age int,\n" +
//                "    primary key(id) not enforced\n" +
//                ") with (\n" +
//                "    'connector' = 'upsert-kafka',\n" +
//                "    'properties.bootstrap.servers' = 'hadoop102:9092',\n" +
//                "    'properties.group.id' = 'tsing',\n" +
//                "    'topic' = 's34',\n" +
//                "    'key.format' = 'json',\n" +
//                "    'value.format' = 'json'\n" +
//                ")"
//        );

        tableEnv.executeSql(
                "create table t34 ( \n" +
                "    id string, \n" +
                "    name string, \n" +
                "    age int\n" +
                ") with ( \n" +
                "    'connector' = 'kafka', \n" +
                "    'properties.bootstrap.servers' = 'hadoop102:9092', \n" +
                "    'properties.group.id' = 'tsing1', \n" +
                "    'topic' = 's34', \n" +
                "    'scan.startup.mode' = 'earliest-offset', \n" +
                "    'format' = 'json'\n" +
                ")"
        );

        tableEnv.sqlQuery("select * from t34").execute().print();
    }
}


